package com.xxd.dfs.namenode;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-07-13 10:48
 * @description: 负责fsimage文件上传的server
 */
public class FSImageReceiveThread extends Thread {
    private final static Logger logger = LoggerFactory.getLogger(FSImageReceiveThread.class);
    private final int PORT = Integer.parseInt(Configuration.getInstance().get("fsimage.port"));
    private final String fsimageDir = Configuration.getInstance().get("editlog.dir");
    private String lastFsimageFileName = "";

    private Selector selector;
    private FSNameSystem nameSystem;

    public FSImageReceiveThread(FSNameSystem nameSystem, String fsimageFileName) {
        this.nameSystem = nameSystem;
        this.lastFsimageFileName = fsimageFileName;
        init();
    }

    /**
     * 初始化
     */
    private void init() {
        logger.info("正在初始化");
        ServerSocketChannel serverSocketChannel = null;

        try {
            selector = Selector.open();
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(PORT), 100);
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        logger.info("FSImageUploadServer启动，监听9000端口......");

        while (true) {
            try {
                selector.select(); // 会阻塞住

                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    keyIterator.remove();
                    handleRequest(key);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 处理监听到的请求
     *
     * @param key
     */
    private void handleRequest(SelectionKey key) throws IOException {
        if (key.isAcceptable()) {
            handleConnectRequest(key);
        } else if (key.isReadable()) {
            handleReadableRequest(key);
        } else if (key.isWritable()) {
            handleWritableRequest(key);
        }
    }

    /**
     * 写数据请求
     *
     * @param key
     */
    private void handleWritableRequest(SelectionKey key) throws IOException {
        logger.info("监听到一个写请求。。。");

        SocketChannel channel = null;

        try {
            channel = (SocketChannel) key.channel();

            ByteBuffer buffer = ByteBuffer.allocate(1024);
            buffer.put("SUCCESS".getBytes());
            buffer.flip();

            channel.write(buffer);

            logger.info("fsimage上传完毕，返回响应SUCCESS给backupnode......");
        } catch (Exception e) {
            e.printStackTrace();
            if (channel != null) {
                channel.close();
            }
        } finally {
            if (key != null) {
                key.cancel();
            }
        }
    }


    /**
     * 读取数据请求,将读取到的数据，写入磁盘文件
     *
     * @param key
     */
    private void handleReadableRequest(SelectionKey key) throws IOException {
        logger.info("接收到读取数据请求");
        SocketChannel channel = null;
        RandomAccessFile fileRAF = null;
        FileOutputStream outputStream = null;
        FileChannel fileChannel = null;
        try {
            // 删除旧文件
            File file = new File(lastFsimageFileName);
            if (file.exists()) {
                file.delete();
            }

            channel = (SocketChannel) key.channel();
            long checkpointTxid = nameSystem.getCheckpointTxid();
            String fsimageFile = fsimageDir + "fsimage_" + checkpointTxid + ".meta";
            lastFsimageFileName = fsimageFile;

            fileRAF = new RandomAccessFile(fsimageFile, "rw");
            outputStream = new FileOutputStream(fileRAF.getFD());
            fileChannel = outputStream.getChannel();

            // 开始读取数据
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int count = 0;
            while ((count = channel.read(buffer)) > 0) {
                buffer.flip();
                fileChannel.write(buffer);
                buffer.clear();
            }

            fileChannel.force(false);

            key.interestOps(SelectionKey.OP_WRITE);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (fileChannel != null) {
                fileChannel.close();
            }
            if (outputStream != null) {
                outputStream.close();
            }
            if (fileRAF != null) {
                fileRAF.close();
            }
        }
    }

    /**
     * 连接请求
     *
     * @param key
     */
    private void handleConnectRequest(SelectionKey key) throws IOException {
        ServerSocketChannel serverSocketChannel = null;
        SocketChannel channel = null;
        try {
            serverSocketChannel = (ServerSocketChannel) key.channel();
            channel = serverSocketChannel.accept();

            logger.info("接收到一个连接请求：" + channel);

            channel.configureBlocking(false);
            channel.register(selector, SelectionKey.OP_READ);
        } catch (Exception e) {
            e.printStackTrace();

            if (channel != null) {
                channel.close();
            }
        }
    }
}

